////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//                                                                                                                            //
//                                                         WNProject                                                          //
//                                                                                                                            //
//            This file is distributed under the BSD 2-Clause open source license. See License.txt for details.               //
//                                                                                                                            //
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

#include "WNNetworking/inc/WNNetworkManagerLinux.h"
#include "WNMemory/inc/WNMemory.h"
#include "WNCore/inc/WNAssert.h"
#include "WNMath/inc/WNBasic.h"
#include "WNConcurrency/inc/WNThread.h"
#include "WNConcurrency/inc/WNSpinMutex.h"
#include "WNNetworking/inc/WNLinuxConnection.h"
#include "WNNetworking/inc/WNLinuxListenConnection.h"
#include "WNNetworking/inc/WNLinuxINConnection.h"
#include "WNNetworking/inc/WNLinuxOutConnection.h"
#include "WNConcurrency/inc/WNLockGuard.h"
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <stdlib.h>
#include <algorithm>

using namespace WNCore;
using namespace WNNetworking;
using namespace WNConcurrency;

WNNetworkManagerLinux::WNNetworkManagerLinux() :
    WNNetworkManager(),
    mInitializationState(eWNNotStarted),
    mWriteEPollInstance(-1), 
    mReadEPollInstance(-1),
    mListenThread(WN_NULL),
    mShuttingDown(WN_FALSE) {
}

WNNetworkManagerLinux::~WNNetworkManagerLinux() {
    Cleanup();
}

WN_VOID WNNetworkManagerLinux::Cleanup() {
    WNNetworkManager::Cleanup();
    mShuttingDown = WN_TRUE;
    if(mInitializationState > eWNEPollListenCreated) {
        close(mListenEPollInstance);
        if (mListenThread != WN_NULL) {
            mListenThread->Join();
            WN_DELETE(mListenThread);
            mListenThread = WN_NULL;
        }
    }

    if(mInitializationState > eWNEPollReadCreated) {
        close(mReadEPollInstance);
        for(WN_SIZE_T i = 0; i < mReadThreads.size(); ++i) {
            mReadThreads[i]->Join();
            WN_DELETE(mReadThreads[i]);
        }
        mReadThreads.clear();
    }

    if(mInitializationState > eWNEPollWriteCreated) {
        close(mWriteEPollInstance);
        for(WN_SIZE_T i = 0; i < mWriteThreads.size(); ++i) {
            mWriteThreads[i]->Join();
            WN_DELETE(mWriteThreads[i]);
        }
        mWriteThreads.clear();
    }
    CleanAllConnections();
}

WNNetworkingError WNNetworkManagerLinux::Initialize(WN_UINT32 _numWorkerThreads) {

    mShuttingDown = WN_FALSE;
    mWriteEPollInstance = epoll_create1(0);
    if (mWriteEPollInstance == -1) {
       Cleanup();
       return(WNNetworking::eWNSystemInitializationFailed);
    }
    
    mInitializationState = eWNEPollWriteCreated;

    for (WN_UINT32 i = 0; i < WNMath::WNCeil(static_cast<WN_FLOAT32>(_numWorkerThreads) / 2.0f); ++i) {
        WNThread* thread = WN_NEW(WNThread(WNWriteThread, this));

        if (!thread->IsValid()) {
            WN_DELETE(thread);
            mInitializationState = eWNWriteThreadsCreated;
            Cleanup();
            return(eWNThreadCreationError);
        }

        mReadThreads.push_back(thread);
    }

    mInitializationState = eWNWriteThreadsCreated;

    mReadEPollInstance = epoll_create1(0);
    if (mReadEPollInstance == -1) {
        Cleanup();

        return(WNNetworking::eWNSystemInitializationFailed);
    }
    
    mInitializationState = eWNEPollReadCreated;

        
    for (WN_UINT32 i = 0; i < WNMath::WNCeil(static_cast<WN_FLOAT32>(_numWorkerThreads) / 2.0f); ++i) {
        WNThread* thread = WN_NEW(WNThread(WNReadThread, this));

        if (!thread->IsValid()) {
            WN_DELETE(thread);
            mInitializationState = eWNReadThreadsCreated;
            Cleanup();
            return(eWNThreadCreationError);
        }

        mReadThreads.push_back(thread);
    }
    mInitializationState = eWNReadThreadsCreated;

    mListenEPollInstance = epoll_create1(0);
    if (mListenEPollInstance == -1) {
        Cleanup();
        return(WNNetworking::eWNSystemInitializationFailed);
    }
    
    mInitializationState = eWNEPollListenCreated;
    
    mListenThread = WN_NEW(WNThread(WNListenThread, this));
    if(!mListenThread->IsValid()) {
        WN_DELETE(mListenThread);
        mListenThread = WN_NULL;
        Cleanup();
        return(eWNThreadCreationError);
    }

    mInitializationState = eWNListenThreadCreated;

    mInitializationState = eWNInitializationComplete;

    return(WNNetworking::eWNOK);
}

WN_VOID WNNetworkManagerLinux::WNListenThread(WNNetworkManagerLinux* _manager) {
    WN_INT32 numEvents;
    static WN_INT32 MAX_EPOLL_EVENTS=1;
    epoll_event events[MAX_EPOLL_EVENTS];
    WNMemory::WNMemSet(events, 0, sizeof(epoll_event) * MAX_EPOLL_EVENTS);
    while(1){
        numEvents = epoll_wait(_manager->mListenEPollInstance, events, MAX_EPOLL_EVENTS, 100);
        
        if(numEvents == -1) {
            char buff[1024];
            strerror_r(errno, buff, 1023);
            printf("%s", buff);
        }
        if(_manager->mShuttingDown) {
            return;
        }
        
        for(WN_INT32 i = 0; i < numEvents; ++i) {
            if((events[i].events & EPOLLERR)) {
                //ERROR;
                continue;
            } else if ((events[i].events & EPOLLHUP)) {
                //internal error?
                continue;
            } else if (!(events[i].events & EPOLLIN)) {
                //UHH whats this?
                continue;
            }
            WNLinuxListenConnection* conn = reinterpret_cast<WNLinuxListenConnection*>(events[i].data.ptr);
            WNLinuxConnection * inConn;
            //Could be getting multiple connections at once
            while((inConn = conn->ReceiveConnection()) != WN_NULL) {
                if(_manager->AddToReadEPoll(inConn) && _manager->AddToWriteEPoll(inConn)) {
                    WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex>(_manager->mIncommingMutex);
                    _manager->mIncommingConnections.push_back(inConn);
                } else {
                    WN_DELETE(inConn);
                }
            }
        }
    }
    
}

WN_VOID WNNetworkManagerLinux::WNReadThread(WNNetworkManagerLinux* _manager) {
    WN_INT32 numEvents;
    static WN_INT32 MAX_EPOLL_EVENTS=1;
    epoll_event events[MAX_EPOLL_EVENTS];
    WNMemory::WNMemSet(events, 0, sizeof(epoll_event) * MAX_EPOLL_EVENTS);
   
    while(1){
        numEvents = epoll_wait(_manager->mReadEPollInstance, events, MAX_EPOLL_EVENTS, 100);
        if(_manager->mShuttingDown) {
            return;
        }
        for(WN_INT32 i = 0; i < numEvents; ++i) {
            if((events[i].events & EPOLLERR)) {
                //ERROR;
                continue;
            } else if ((events[i].events & EPOLLHUP)) {
                //internal error?
                continue;
            } else if (!(events[i].events & EPOLLIN)) {
                //UHH whats this?
                continue;
            }
            WNLinuxConnection* conn = reinterpret_cast<WNLinuxConnection*>(events[i].data.ptr);
            conn->NotifyReadReady();
        }
    }
}

WN_VOID WNNetworkManagerLinux::WNWriteThread(WNNetworkManagerLinux* _manager) {
    WN_INT32 numEvents;
    static WN_INT32 MAX_EPOLL_EVENTS=1;
    static WN_INT32 MAX_EPOLL_TIME=5;
    epoll_event events[MAX_EPOLL_EVENTS];
   
    while(1){
        numEvents = epoll_wait(_manager->mWriteEPollInstance, events, MAX_EPOLL_EVENTS, 100);
        if(_manager->mShuttingDown) {
            return;
        }
        for(WN_INT32 i = 0; i < numEvents; ++i) {
            if((events[i].events & EPOLLERR) ||
               (events[i].events & EPOLLHUP) ||
               !((events[i].events & EPOLLOUT))) {
                //HANDLE CLOSING CONNECTION
                continue;
            }
            WNLinuxConnection* conn = reinterpret_cast<WNLinuxConnection*>(events[i].data.ptr);
            conn->NotifyReadyToSend(WN_TRUE);
        }
    }
}

WNNetworkingError WNNetworkManagerLinux::CreateListener(WNConnection*& _outHandle, WNConnectionType _type, WN_UINT16 _port, const WNConnectedCallback& _callback) {
    WNLinuxListenConnection* conn = WN_NEW(WNLinuxListenConnection(*this, _type, _port, _callback));
    WNNetworkingError err = conn->Initialize();
    
    if (err != eWNOK) {
        WN_DELETE(conn);
        return(err);
    }

    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = conn;
    event.events = EPOLLIN | EPOLLET;
    
    if(epoll_ctl(mListenEPollInstance, EPOLL_CTL_ADD, conn->GetLinuxSocket(), &event) != 0) {
        WN_DELETE(conn);
        return(WNNetworking::eWNCannotAssociate);
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mListenMutex);
        mListenConnections.push_back(conn);
    }
    _outHandle = conn;

    return(eWNOK);
}

WN_BOOL WNNetworkManagerLinux::AddToReadEPoll(WNLinuxConnection* _conn) {
    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = _conn;
    event.events = EPOLLIN | EPOLLET;
    if(epoll_ctl(mReadEPollInstance, EPOLL_CTL_ADD, _conn->GetLinuxSocket(), &event) == -1) {
        return(WN_FALSE);
    }
    return(WN_TRUE);
}

WN_BOOL WNNetworkManagerLinux::AddToWriteEPoll(WNLinuxConnection* _conn) {
    struct epoll_event event;
    WNMemory::WNMemSet(&event, 0, sizeof(epoll_event));
    event.data.ptr = _conn;
    event.events = EPOLLOUT | EPOLLET;
    if(epoll_ctl(mWriteEPollInstance, EPOLL_CTL_ADD, _conn->GetLinuxSocket(), &event) == -1) {
        return(WN_FALSE);
    }
    return(WN_TRUE);
}

WNNetworkingError WNNetworkManagerLinux::ConnectTo(WNConnection*& _outHandle, WNConnectionType _type, const WN_CHAR* _target, WN_UINT16 _port) {
    WN_RELEASE_ASSERT_DESC(_type == eWNReliable, "WNNetworking does not support unreliable connections .... YET");
    WNLinuxOutConnection* conn = WN_NEW(WNLinuxOutConnection(*this));
    WNNetworkingError err = conn->Initialize(_type, _target, _port);
    if(err != eWNOK) {
        WN_DELETE(conn);
        return(err);
    }
    if(!AddToReadEPoll(conn)) {
        WN_DELETE(conn);
        return(WNNetworking::eWNCannotAssociate);
    }
    if(!AddToWriteEPoll(conn)) {
        WN_DELETE(conn);
        return(WNNetworking::eWNCannotAssociate);
    }
    {
        mOutgoingMutex.Lock();
        mOutgoingConnections.push_back(conn);
        mOutgoingMutex.Unlock();
    }
    _outHandle = conn;
    return(WNNetworking::eWNOK);
}

WN_VOID WNNetworkManagerLinux::DestroyConnection(WNConnection* _connection) {
     {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mIncommingMutex);
        std::list<WNLinuxConnection*>::iterator i = std::find(mIncommingConnections.begin(), mIncommingConnections.end(), _connection);
        if(i != mIncommingConnections.end()) {
             (*i)->Invalidate();
            mIncommingConnections.erase(i);
        } 
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mOutgoingMutex);
        std::list<WNLinuxConnection*>::iterator i = std::find(mOutgoingConnections.begin(), mOutgoingConnections.end(), _connection);
        if(i != mOutgoingConnections.end()) {
             (*i)->Invalidate();
            mOutgoingConnections.erase(i);
        } 
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mListenMutex);
        std::list<WNLinuxListenConnection*>::iterator i = std::find(mListenConnections.begin(), mListenConnections.end(), _connection);
        if(i != mListenConnections.end()) {
             (*i)->Invalidate();
            mListenConnections.erase(i);
        }
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mInvalidMutex);
        mInvalidConnections.push_back(_connection);
    }
}

WN_VOID WNNetworkManagerLinux::CleanAllConnections() {
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mIncommingMutex);
        for(std::list<WNLinuxConnection*>::iterator i = mIncommingConnections.begin(); i != mIncommingConnections.end(); ++i) {
            WN_DELETE((*i));
        } 
        mIncommingConnections.clear();
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mOutgoingMutex);
        for(std::list<WNLinuxConnection*>::iterator i = mOutgoingConnections.begin(); i != mOutgoingConnections.end(); ++i) {
            WN_DELETE((*i));
        } 
        mOutgoingConnections.clear();
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mListenMutex);
        for(std::list<WNLinuxListenConnection*>::iterator i = mListenConnections.begin(); i != mListenConnections.end(); ++i) {
            WN_DELETE((*i));
        } 
        mListenConnections.clear();
    }
    {
        WNConcurrency::WNLockGuard<WNConcurrency::WNSpinMutex> guard(mInvalidMutex);
        mInvalidConnections.clear();
    }
}
